package com.lagou.rpc.consumer.client;

import com.lagou.rpc.consumer.handler.RpcClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 客户端
 * 1.连接netty服务端
 * 2、提供给调用者主动关闭连接
 * 3、提供消息发送的方法
 */
@Component
public class RpcClient {

    private String ip;

    private int port;

    EventLoopGroup group;
    Bootstrap bootstrap;
    Channel channel;

    @Autowired
    private RpcClientHandler rpcClientHandler ;

    ExecutorService executorService = new ThreadPoolExecutor(10,30,3000,
            TimeUnit.SECONDS,new ArrayBlockingQueue<>(1000));

//    public RpcClient(String ip, int port) {
//        this.ip = ip;
//        this.port = port;
//        initClient();
//    }

    public void initClient(String ip,int port) {
        try {

            //创建线程组
            group = new NioEventLoopGroup();
            bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            ChannelPipeline pipeline = channel.pipeline();
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            //添加客户端处理类
                            pipeline.addLast(rpcClientHandler);

                        }
                    });
            channel = bootstrap.connect(ip, port).sync().channel();
        } catch (Exception e) {
            e.printStackTrace();
            if (group != null) {
                group.shutdownGracefully();
            }

            if (channel != null) {
                channel.close();
            }
        }


    }

    public void close() {
        if (group != null) {
            group.shutdownGracefully();
        }

        if (channel != null) {
            try {
                channel.closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public Object send(String msg) throws ExecutionException, InterruptedException {
        rpcClientHandler.setRequestMsg(msg);
        System.out.println("ddddddddddddddddd");
        Future future = executorService.submit(rpcClientHandler);
        return future.get();
    }

}
